package com.to8to.kitt.esb.http;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.curator.x.discovery.ServiceInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Splitter;
import com.to8to.kitt.Constants;
import com.to8to.kitt.ThriftBizServerChannelInitializer;
import com.to8to.kitt.ThriftClassFactory;
import com.to8to.kitt.esb.AbstractServerDeamon;
import com.to8to.kitt.esb.BusServer;
import com.to8to.kitt.esb.NodeChangeEvent;
import com.to8to.kitt.esb.ZKServiceChannelPool;
import com.to8to.kitt.utils.DateUtil;
import com.to8to.kitt.utils.LogUtils;

public class HttpBusServer extends AbstractServerDeamon{

    public static Logger logger = LoggerFactory.getLogger(HttpBusServer.class);

    private static Map<String, HttpBusMessageState> StateMap = new HashMap<String, HttpBusMessageState>();
    
    private static ReadWriteLock lock = new ReentrantReadWriteLock();     
    
    int dumpWaitMs = 5000;
    
    ZKServiceChannelPool pool;
    
    public static Map<String, String> nsMap = new HashMap<String, String>();
    
    public static void putState(String key, HttpBusMessageState state)
	{
    	lock.writeLock().lock();
		try
		{
			StateMap.put(key, state);
		}
		catch(Exception ee)
		{
			
		}
		finally
		{
			lock.writeLock().unlock();
		}
	}
	
	
    public static HttpBusMessageState getAndRemoveState(String key)
	{
    	lock.writeLock().lock();
		try
		{
			HttpBusMessageState v = StateMap.get(key);
			StateMap.remove(key);
			return v;
		}
		catch(Exception ee)
		{
			
		}
		finally
		{
			lock.writeLock().unlock();
		}
		return null;
	}
    
    
    public static void cleanOnce()
    {
    	lock.writeLock().lock();
    	try
		{
    		Set<String> keyset = StateMap.keySet();
    		long now = System.currentTimeMillis();
    		for(String key: keyset)
    		{
    			HttpBusMessageState status = StateMap.get(key);
    			if(status == null)
    			{
    				StateMap.remove(key);
    			}
    			else
    			{
    				if(now - status.getRecvFromClientTime() > Constants.ESB_EXP_TIME_MS) //timeout
    				{
    					logger.trace("清理超时的任务: " + status.getName() + ", 收到请求时间: " + DateUtil.format(new Date(status.getRecvFromClientTime())));
    					
    					status = null;
    					
    					StateMap.remove(key);
    				}
    			}
    		}
		}
		catch(Exception ee)
		{
			
		}
		finally
		{
			lock.writeLock().unlock();
		}
    }
	
    
	@Override
	public void onPrivateConfigChange(Map<String, NodeChangeEvent> config) {

		String[] files = {/*"ns.conf", */"jars.conf" , "server.properties"};
		
		for(String file: files)
		{
			NodeChangeEvent event = config.get(file);
			
			if (event != null && event.isChange()) 
			{
				System.out.println(event);
				
				byte[] data = event.ndata;
				
				if(file.equals("server.properties"))
				{
					try {
						
						PropertiesConfiguration pc = new PropertiesConfiguration();
						
						pc.load(new ByteArrayInputStream(data));
						
						String bind_host = pc.getString("bind_host");
						
						int bind_port = pc.getInt("http_port");
											
						
						boolean rebind = false;
						if(!bind_host.equals(this.bindHost) || this.bindPort != bind_port)
							rebind = true;
						 
						this.bindHost = bind_host;
						this.bindPort = bind_port;
						 
						if(rebind)
							bind();
						
					} catch (ConfigurationException e) {
						logger.error(LogUtils.logStackTrace(e));
					}
				}
				else if(file.equals("jars.conf"))
				{
					try {
						
						String txt = new String(data);
						
						List<String> jars = Splitter.on("\n").trimResults().omitEmptyStrings().splitToList(txt);
						
						URL[] urls = new URL[jars.size()];
						int i = 0;
						for(String jar : jars)
						{
							URL url = new File(jar).toURI().toURL();	
							urls[i++] = url;
							logger.info("加载jar包: " + jar);
						}
						
						ThriftClassFactory.classLoader = new URLClassLoader(urls, Thread.currentThread().getContextClassLoader());  
						
					} catch (Exception ee) {
						// TODO Auto-generated catch block
						ee.printStackTrace();
					}
				}
			}
		}
	}
    
	@Override
	public void onCommonConfigChange(NodeChangeEvent e) {
		
		if(e.isChange())
		{
			if(e.path.contains("ns.conf"))
			{
				try {
					String txt = new String(e.ndata);
					List<String> lines = Splitter.on("\n").trimResults().omitEmptyStrings().splitToList(txt);
					for(String line : lines)
					{
						List<String> parts = Splitter.on(" ").splitToList(line);
						if(parts.size()==2)
						{
							String ns = parts.get(1);
							String classname = parts.get(0);
							logger.info("加载命名空间 =======> ns:{}, class:{}", ns, classname);
							nsMap.put(ns, classname);
						}
					}
				} catch (Exception ee) {
					// TODO Auto-generated catch block
					ee.printStackTrace();
				}
			}
		}
		
	}
	
	public void start(String instance)
	{
		Http2ThriftHandler http2thriftHandler = new Http2ThriftHandler();
		
		Thrift2HttpHandler thrift2httpHandler = new Thrift2HttpHandler();
		
		HttpServerInitializer serverChannelInitializer = new HttpServerInitializer(http2thriftHandler);
		
		ThriftBizServerChannelInitializer clientChannelInitilizer = new ThriftBizServerChannelInitializer(thrift2httpHandler);
		
		Bootstrap clientBootstap = new Bootstrap();
		
        clientBootstap.group(new NioEventLoopGroup(1))
        .channel(NioSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY, true)
        .option(ChannelOption.SO_KEEPALIVE, true)
        .handler(clientChannelInitilizer);
		
        pool = new ZKServiceChannelPool(clientBootstap);
        
		http2thriftHandler.setServiceChannelPool(pool);

        setChannelInitilizer(serverChannelInitializer);
		
		setServiceInstance(BusServer.class.getName(), instance);
		
		setReportAddressToZK(false);
		
		setSyncConfigFromZK(true);
		
		super.addCommonConfigPath("crm/common/ns.conf");
//		super.addCommonConfigPath("/crm/common/jars.conf");
		
		pool.init();
		
		if(!init())
		{
			logger.error("server init failed!");
			System.exit(-1);
		}
		
//		deamon();
	}
	
	public void deamon()
	{
		Thread t = new Thread(new Runnable(){

			@Override
			public void run() {
				while(true)
				{
					try {
						String newDump = pool.dumpServiceConnectionsMT();
						String oldDump = pool.getServiceInstancePayload(service, instance);
						if(!newDump.equals(oldDump))
						{
							logger.info("dump changed!!!! flush ESB-down connection dumps!!!");
							serviceInstance = ServiceInstance.<String>builder().id(instance).name(service).address(reportHost).payload(newDump).port(reportPort).build();
							serviceDiscovery.updateService(serviceInstance);
						}
					} catch (Exception e1) {
						e1.printStackTrace();
					}
					
					try {
						Thread.sleep(dumpWaitMs);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
			
		});
		
		t.start();
	}
	
	public static void main(String[] args){
		
		String instance = "1";
		
		if(args.length > 0)
			instance = args[0];
		
		HttpBusServer server = new HttpBusServer();
		
		server.start(instance);
	}


}
